 1.Kafka高级特性解析之消费者中消费接收下的再均衡
   
   重平衡可以说是kafka为人诟病最多的一个点了。
   重平衡其实就是一个协议，它规定了如何让消费者组下的所有消费者来分配topic中的每一个分区。比如一个topic
有100个分区，一个消费者组内有20个消费者，在协调者的控制下让组内每一个消费者分配到5个分区，这个分配的过
程就是重平衡。
   重平衡的触发条件主要有三个：
   1).消费者组内成员发生变更，这个变更包括了增加和减少消费者，比如消费者宕机退出消费组。
   2).主题的分区数发生变更，kafka目前只支持增加分区，当增加的时候就会触发重平衡
   3).订阅的主题发生变化，当消费者组使用正则表达式订阅主题，而恰好又新建了对应的主题，就会触发重平衡
   消费者宕机，退出消费组，触发再平衡，重新给消费组中的消费者分配分区。
   由于broker宕机，主题X的分区3宕机，此时分区3没有Leader副本，触发再平衡，消费者4没有对应的主题分区，
则消费者4闲置。
   主题增加分区，需要主题分区和消费组进行再均衡。
   由于使用正则表达式订阅主题，当增加的主题匹配正则表达式的时候，也要进行再均衡。
   为什么说重平衡为人诟病呢？因为重平衡过程中，消费者无法从kafka消费消息，这对kafka的TPS影响极大，而
如果kafka集内节点较多，比如数百个，那重平衡可能会耗时极多。数分钟到数小时都有可能，而这段时间kafka基本
处于不可用状态。所以在实际环境中，应该尽量避免重平衡发生。
 
 2.避免重平衡
   
   要说完全避免重平衡，是不可能，因为你无法完全保证消费者不会故障。⽽消费者故障其实也是最常见的引发重
平衡的地方，所以我们需要保证尽力避免消费者故障。
   而其他几种触发重平衡的方式，增加分区，或是增加订阅的主题，抑或是增加消费者，更多的是主动控制。
   如果消费者真正挂掉了，就没办法了，但实际中，会有一些情况，kafka错误地认为一个正常的消费者已经挂掉
了，我们要的就是避免这样的情况出现。
   首先要知道哪些情况会出现错误判断挂掉的情况。
   在分布式系统中，通常是通过心跳来维持分布式系统的，kafka也不例外。
   在分布式系统中，由于网络问题你不清楚没接收到心跳，是因为对方真正挂了还是只是因为负载过重没来得及发
生心跳或是网络堵塞。所以一般会约定一个时间，超时即判定对方挂了。而在kafka消费者场景中,session.timout.ms
参数就是规定这个超时时间是多少。
   还有一个参数，heartbeat.interval.ms，这个参数控制发送心跳的频率，频率越高越不容易被误判，但也会消
耗更多资源。
   此外，还有最后一个参数，max.poll.interval.ms，消费者poll数据后，需要一些处理，再进行拉取。如果两次
拉取时间间隔超过这个参数设置的值，那么消费者就会被踢出消费者组。也就是说，拉取，然后处理，这个处理的时
间不能超过max.poll.interval.ms 这个参数的值。这个参数的默认值是5分钟，而如果消费者接收到数据后会执行
耗时的操作，则应该将其设置得大一些。
   三个参数，
   session.timout.ms控制心跳超时时间，
   heartbeat.interval.ms控制心跳发送频率，
   max.poll.interval.ms控制poll的间隔。
   这里给出一个相对较为合理的配置，如下：
       session.timout.ms：设置为6s
       heartbeat.interval.ms：设置2s
       max.poll.interval.ms：推荐为消费者处理消息最长耗时再加1分钟
 
 3.消费者拦截器
   
   消费者在拉取了分区消息之后，要首先经过反序列化器对key和value进行反序列化处理。
   处理完之后，如果消费端设置了拦截器，则需要经过拦截器的处理之后，才能返回给消费者应用程序进行处理。
   消费端定义消息拦截器，需要实现org.apache.kafka.clients.consumer.ConsumerInterceptor<K, V> 接口。
   1).一个可插拔接口，允许拦截甚至更改消费者接收到的消息。⾸要的⽤例在于将第三方组件引入消费者应用程
序，用于定制的监控、日志处理等。
   2).该接口的实现类通过configre方法获取消费者配置的属性，如果消费者配置中没有指定clientID，还可以获取
KafkaConsumer生成的clientId。获取的这个配置是跟其他拦截器共享的，需要保证不会在各个拦截器之间产生冲突。
   3).ConsumerInterceptor方法抛出的异常会被捕获、记录，但是不会向下传播。如果用户配置了错误的key或
value类型参数，消费者不会抛出异常，而仅仅是记录下来。
   4).ConsumerInterceptor回调发⽣在org.apache.kafka.clients.consumer.KafkaConsumer#poll(long)方法同
一个线程。
   该接口中有如下方法：
package org.apache.kafka.clients.consumer;
import org.apache.kafka.common.Configurable;
import org.apache.kafka.common.TopicPartition;
import java.util.Map;
public interface ConsumerInterceptor<K, V> extends Configurable {
/**
*
* 该方法在poll方法返回之前调口。调用结束后poll方法就返回消息了。
*
* 该⽅法可以修改消费者消息，返回新的消息。拦截器可以过滤收到的消息或⽣成新的消息。
* 如果有多个拦截器，则该方法按照KafkaConsumer的configs中配置的顺序调⽤。
*
* @param records 由上个拦截器返回的由客户端消费的消息。
*/
    public ConsumerRecords<K, V> onConsume(ConsumerRecords<K, V> records);
/**
* 当消费者提交偏移量时，调用该方法。
* 该方法抛出的任何异常调用者都会忽略。
*/
    public void onCommit(Map<TopicPartition, OffsetAndMetadata> offsets);
    public void close();
}
   代码实现
   com.lagou.kafka.demo.consumer.MyConsumer
package com.lagou.kafka.demo.consumer;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.util.Collections;
import java.util.Properties;

public class MyConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "linux121:9092");
        props.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        props.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        props.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "mygrp");
//        props.setProperty(ConsumerConfig.CLIENT_ID_CONFIG, "myclient");
        // 如果在kafka中找不到当前消费者的偏移量，则设置为最旧的
        props.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        // 配置拦截器
        // One -> Two -> Three,接收消息和发送消息偏移量确认都是这个顺序
        props.setProperty(ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG,
                "com.lagou.kafka.demo.interceptor.OneInterceptor" +
                ",com.lagou.kafka.demo.interceptor.TwoInterceptor" +
                ",com.lagou.kafka.demo.interceptor.ThreeInterceptor");
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
        //订阅主题
        consumer.subscribe(Collections.singleton("tp_demo_01"));
        while (true) {
            final ConsumerRecords<String, String> records = consumer.poll(3_000);
            records.forEach(record -> {
                System.out.println(record.topic()
                   + "\t" + record.partition()
                   + "\t" + record.offset()
                   + "\t" + record.key()
                   + "\t" + record.value());
            });
//            consumer.commitAsync();
//            consumer.commitSync();
        }
//        consumer.close();

    }
}


   com.lagou.kafka.demo.interceptor.OneInterceptor
package com.lagou.kafka.demo.interceptor;

import org.apache.kafka.clients.consumer.ConsumerInterceptor;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;

import java.util.Map;

public class OneInterceptor implements ConsumerInterceptor<String, String> {
    @Override
    public ConsumerRecords<String, String> onConsume(ConsumerRecords<String, String> records) {
        // poll方法返回结果之前最后要调用的方法
        System.out.println("One -- 开始");
        //消息不做处理，直接返回
        return records;
    }

    @Override
    public void onCommit(Map<TopicPartition, OffsetAndMetadata> offsets) {
          // 消费者提交偏移量的时候，经过该方法
        System.out.println("One -- 结束");
    }

    @Override
    public void close() {
        //用于关闭拦截器用到的资源，如打开的文件，连接的数据库等。
    }

    @Override
    public void configure(Map<String, ?> configs) {
        // 用于获取消费者的设置参数
        configs.forEach((k, v) -> {
            System.out.println(k + "\t" + v);
        });

    }
}


   com.lagou.kafka.demo.interceptor.TwoInterceptor
package com.lagou.kafka.demo.interceptor;

import org.apache.kafka.clients.consumer.ConsumerInterceptor;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;

import java.util.Map;

public class TwoInterceptor implements ConsumerInterceptor<String, String> {
    @Override
    public ConsumerRecords<String, String> onConsume(ConsumerRecords<String, String> records) {
        // poll方法返回结果之前最后要调用的方法
        System.out.println("Two -- 开始");
        //消息不做处理，直接返回
        return records;
    }

    @Override
    public void onCommit(Map<TopicPartition, OffsetAndMetadata> offsets) {
          // 消费者提交偏移量的时候，经过该方法
        System.out.println("Two -- 结束");
    }

    @Override
    public void close() {
        //用于关闭拦截器用到的资源，如打开的文件，连接的数据库等。
    }

    @Override
    public void configure(Map<String, ?> configs) {
        // 用于获取消费者的设置参数
        configs.forEach((k, v) -> {
            System.out.println(k + "\t" + v);
        });

    }
}


   com.lagou.kafka.demo.interceptor.ThreeInterceptor		
package com.lagou.kafka.demo.interceptor;

import org.apache.kafka.clients.consumer.ConsumerInterceptor;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;

import java.util.Map;

public class ThreeInterceptor implements ConsumerInterceptor<String, String> {
    @Override
    public ConsumerRecords<String, String> onConsume(ConsumerRecords<String, String> records) {
        // poll方法返回结果之前最后要调用的方法
        System.out.println("Three -- 开始");
        //消息不做处理，直接返回
        return records;
    }

    @Override
    public void onCommit(Map<TopicPartition, OffsetAndMetadata> offsets) {
          // 消费者提交偏移量的时候，经过该方法
        System.out.println("Three -- 结束");
    }

    @Override
    public void close() {
        //用于关闭拦截器用到的资源，如打开的文件，连接的数据库等。
    }

    @Override
    public void configure(Map<String, ?> configs) {
        // 用于获取消费者的设置参数
        configs.forEach((k, v) -> {
            System.out.println(k + "\t" + v);
        });

    }
}
	
	
	